package org.kafka;

import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaClientTest {

    public static void main(String [] args){
        KafkaProducer<String,String> producer = getClient();
        ExecutorService service = Executors.newFixedThreadPool(10);

        for (int i = 0;i<100000;i++){
            service.submit(new Runnable() {
                @Override
                public void run() {
                    ProducerRecord<String ,String> record = new ProducerRecord<>("iot-device-property-report",
                            "{\"devId\":\"358800006307525\",\"msgType\":\"report_prop\",\"msgData\":[{\"iid\":\"A.3.D\",\"value\":\"358800006307525\",\"time\":1711952573000}],\"msgTime\":1711952575793}");
                    producer.send(record);
                }
            });
        }
    }

    public static KafkaProducer<String,String> getClient(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.21.32.30:9092");
        properties.put(ProducerConfig.RETRIES_CONFIG, 1);
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 100);
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 10000);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new KafkaProducer<String,String>(properties);
    }

//    public static KafkaConsumer<String,String> getConsumerClient(){
//        Properties properties = new Properties();
//        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,  "172.21.32.30:9092");
//        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);
//        properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1000);
//        properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 1000);
//        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 1000);
//
//        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
//
//        consumerPropertiesPerTopic
//                .getOrDefault(topic, Collections.emptyList())
//                .forEach(kv -> props.put(kv.getKey(), kv.getValue()));
//        return new KafkaConsumer<String,String>(properties);
//    }
}
